package hust.rpc;

import hust.rpc.annotation.Service;
import hust.rpc.constants.Constants;
import hust.rpc.handler.*;
import hust.rpc.invoker.RemoteInvoker;
import hust.rpc.lb.LoadBalance;
import hust.rpc.lb.RandomLoadBalance;
import hust.rpc.pojo.Host;
import hust.rpc.proxy.AllProxy;
import hust.rpc.proxy.Proxy;
import hust.rpc.utils.ServiceScanUtils;
import hust.rpc.utils.SingletonUtils;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Set;
import java.util.concurrent.CountDownLatch;

/**
 * 提供者启动器
 */
@Slf4j
public class ProviderBootstrap {

    // 闭锁,用于同步(成功连接注册中心)
    private static final CountDownLatch countDownLatch = new CountDownLatch(1);

    // 共用事件线程池
    private static final EventLoopGroup workerGroup = new NioEventLoopGroup();

    // 本地监听地址
    private static Host host;

    public static Host getHost() {
        return host;
    }

    /**
     * 启动服务提供者连接注册中心、监听本地端口以响应服务消费者
     *
     * @param registryIp   注册中心ip地址
     * @param registryPort 注册中心端口号
     * @param localIp      本地监听ip地址
     * @param localPort    本地监听端口号
     */
    public static void run(String registryIp, int registryPort, String localIp, int localPort) throws InterruptedException, IOException, ClassNotFoundException {
        // 绑定服务
        bindService();
        // 设置负载均衡策略
        setLoadBalance(RandomLoadBalance.class);
        // 设置本地监听信息
        host = new Host(localIp, localPort);
        // 初始化RemoteInvoker
        SingletonUtils.getInstance(RemoteInvoker.class);
        // 连接注册中心
        connectRegistry(registryIp, registryPort);
        // 监听本地端口
        listen(localIp, localPort);
    }

    /**
     * 设置负载均衡策略
     */
    private static void setLoadBalance(Class<? extends LoadBalance> clazz) {
        SingletonUtils.registerInstance(LoadBalance.class, clazz);
    }

    /**
     * 绑定服务
     * 本地服务: 编写的服务接口实现类
     *
     * @see hust.rpc.service.ProviderService
     * 远程服务: 向注册中心注册服务、取消注册服务
     */
    public static void bindService() throws IOException, ClassNotFoundException {
        // 创建代理对象
        SingletonUtils.registerInstance(Proxy.class, AllProxy.class);
        // 扫描包获取@Service标注的类
        Set<Class<?>> services = ServiceScanUtils.scanService(Constants.BASE_SCAN_PACKAGE, (clazz) -> {
            Service annotation = clazz.getAnnotation(Service.class);
            // value为服务提供者 || value为空
            return Constants.PROVIDER.equals(annotation.value()) || annotation.value().isEmpty();
        });
        // 初始化代理对象
        SingletonUtils.getInstance(Proxy.class).initProxy(services);
    }

    /**
     * 闭锁打开
     */
    public static void countDown() {
        countDownLatch.countDown();
    }

    /**
     * 连接注册中心
     *
     * @param registryIp   注册中心ip
     * @param registryPort 注册中心端口
     */
    public static void connectRegistry(String registryIp, int registryPort) throws InterruptedException {
        // Netty公式
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(workerGroup)
                .channel(NioSocketChannel.class)
                // 关闭TCP的默认开启的Nagle算法
                .option(ChannelOption.TCP_NODELAY, true)
                // 开启TCP底层心跳机制
                .option(ChannelOption.SO_KEEPALIVE, true)
                // 设置连接超时时间
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        pipeline.addLast(new MessageEncoder());
                        pipeline.addLast(new MessageDecoder());
                        pipeline.addLast(new RegistryHandler());
                    }
                });
        log.info("服务提供者开始连接注册中心~~~");
        bootstrap.connect(registryIp, registryPort);

        // 等待同步
        countDownLatch.await();
    }

    /**
     * 监听本地端口
     *
     * @param localIp   本地监听ip
     * @param localPort 本地监听端口
     */
    public static void listen(String localIp, int localPort) {
        // Netty公式
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    // 关闭TCP的默认开启的Nagle算法
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    // 开启TCP底层心跳机制
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    // 设置连接超时时间
                    .childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
                    .handler(new LoggingHandler())
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            pipeline.addLast(new MessageEncoder());
                            pipeline.addLast(new MessageDecoder());
                            pipeline.addLast(new RpcInvokeHandler());
                        }
                    });
            ChannelFuture channelFuture = serverBootstrap.bind(localIp, localPort);
            log.info("服务提供者开始运行~~~");
            channelFuture.channel().closeFuture().sync();
        } catch (Exception e) {
            log.error("服务提供者启动失败: ", e);
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}
